package com.thiswhyme.redis.geo;

import com.thiswhyme.redis.geo.entity.GeoMember;
import lombok.AllArgsConstructor;
import org.springframework.data.geo.Circle;
import org.springframework.data.geo.Distance;
import org.springframework.data.geo.GeoResults;
import org.springframework.data.geo.Metrics;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * Geo 模版方法 - spring data redis 实现
 *
 * @author thiswhyme
 * @date 2022/11/23
 */
@AllArgsConstructor
public class RedisDataGeoTemplate extends AbstractGeoTemplate {

    private final RedisTemplate redisTemplate;

    @Override
    public <T> Long add(String key, GeoMember<T> geoMember) {
        return redisTemplate.opsForGeo().add(key, geoMemberToGeoLocation(geoMember));
    }

    @Override
    public <T> Long batchAdd(String key, List<GeoMember<T>> geoMembers) {
        if (CollectionUtils.isEmpty(geoMembers)) {
            return 0L;
        }
        return redisTemplate.opsForGeo().add(key, geoMembersToGeoLocations(geoMembers));
    }

    @Override
    public <T> boolean remove(String key, T member) {
        Long result = redisTemplate.opsForGeo().remove(key, member);
        return Optional.ofNullable(result).map(r -> r > 0).orElse(false);
    }

    @Override
    public boolean clear(String key) {
        return redisTemplate.delete(key);
    }

    @Override
    public <T> Double distance(String key, T fromMember, T toMember, Metrics metric) {
        Distance distance = redisTemplate.opsForGeo().distance(key, fromMember, toMember, metric);
        return Optional.ofNullable(distance).map(Distance::getValue).orElse(null);
    }

    @Override
    public <T> Map<T, Double> radius(String key, double lng, double lat, double radius, Metrics metric, Boolean asc, Integer count) {
        Circle circle = new Circle(new Point(lng, lat), new Distance(radius, metric));
        RedisGeoCommands.GeoRadiusCommandArgs args = getGeoRadiusCommandArgs(asc, count);
        GeoResults<RedisGeoCommands.GeoLocation<T>> results = redisTemplate.opsForGeo().radius(key, circle, args);
        if (null != results && !CollectionUtils.isEmpty(results.getContent())) {
            return results.getContent().stream().collect(Collectors.toMap(n -> n.getContent().getName(), n -> n.getDistance().getValue()));
        }
        return Collections.emptyMap();
    }

    @Override
    public <T> Map<T, Double> radius(String key, T member, double radius, Metrics metric, Boolean asc, Integer count) {
        RedisGeoCommands.GeoRadiusCommandArgs args = getGeoRadiusCommandArgs(asc, count);
        GeoResults<RedisGeoCommands.GeoLocation<T>> results = redisTemplate.opsForGeo().radius(key, member, new Distance(radius, metric), args);
        if (null != results && !CollectionUtils.isEmpty(results.getContent())) {
            return results.getContent().stream().collect(Collectors.toMap(n -> n.getContent().getName(), n -> n.getDistance().getValue()));
        }
        return Collections.emptyMap();
    }

    /**
     * 组装 GeoRadiusCommandArgs
     *
     * @param asc
     * @param count
     * @return
     */
    private RedisGeoCommands.GeoRadiusCommandArgs getGeoRadiusCommandArgs(Boolean asc, Integer count) {
        RedisGeoCommands.GeoRadiusCommandArgs args = RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs()
                .includeCoordinates()
                .includeDistance();
        if (null != asc) {
            if (asc) {
                args.sortAscending();
            } else {
                args.sortDescending();
            }
        }
        if (null != count) {
            args.limit(count);
        }
        return args;
    }

    /**
     * GeoMember 转 GeoLocation
     *
     * @param geoMember
     * @return
     * @param <T>
     */
    private <T> RedisGeoCommands.GeoLocation<T> geoMemberToGeoLocation(GeoMember<T> geoMember) {
        Point point = new Point(geoMember.getLongitude(), geoMember.getLatitude());
        return new RedisGeoCommands.GeoLocation<>(geoMember.getMember(), point);
    }

    /**
     * GeoMember 列表转 GeoLocation 列表
     *
     * @param geoMembers
     * @return
     * @param <T>
     */
    private <T> List<RedisGeoCommands.GeoLocation<T>> geoMembersToGeoLocations(List<GeoMember<T>> geoMembers) {
        List<RedisGeoCommands.GeoLocation<T>> results = new ArrayList<>();
        geoMembers.forEach(geoMember -> results.add(geoMemberToGeoLocation(geoMember)));
        return results;
    }
}
